package com.wish.system.kafka;

import cn.hutool.core.util.RandomUtil;
import com.wish.common.core.config.SmsConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class KafkaListener {

    @Autowired
    RedisTemplate redisTemplate;

    @org.springframework.kafka.annotation.KafkaListener(topics = {"wish"},groupId = "good-group")
    public void test(ConsumerRecord<String,String> record, Acknowledgment acknowledgment) throws Exception {
        if (!redisTemplate.opsForValue().setIfAbsent(record.key(),null,2, TimeUnit.MINUTES)) {
            return;
        }
        String code = RandomUtil.randomNumbers(4);
        redisTemplate.opsForValue().set("code:"+record.value(),code,2, TimeUnit.MINUTES);
        String result = SmsConfig.send(record.value(), code);
        System.err.println(result);
        acknowledgment.acknowledge();
    }
}
